/*
 * Copyright (C) 2012-2013  Romain DUBOIS
 * 
 * This file is part of NeMoS - Network and Modular Software
 * 
 * NeMoS is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * NeMoS is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with NeMoS.  If not, see <http://www.gnu.org/licenses/>.
 */
package nemos.remote.rsa.message.transport;

import java.io.IOException;
import java.rmi.RemoteException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import nemos.remote.rsa.internal.RsaMessage;
import nemos.remote.rsa.message.data.InvocationMessage;
import nemos.remote.rsa.message.data.Message;
import nemos.remote.rsa.message.data.ResponseMessage;
import nemos.util.BlockingReference;
import nemos.util.log.NemosLogger;

/**
 * AbstractTransport : The transport abstract implementation.
 * 
 * @author Romain DUBOIS
 */
public abstract class AbstractTransport implements ITransport {

    /** The incoming messages */
    private final BlockingQueue<Message> mIncomingMessages;
    /** The invocation call results */
    private final Map<Long, BlockingReference<ResponseMessage>> mInvocationResults;
    /** The last sended message ID */
    private final AtomicLong mLastId;
    /** The logger */
    private final NemosLogger mLogger;
    /** The active flag */
    private boolean mActive;
    /** The transport ID */
    private String mId;
    /** The transport label */
    private String mLabel;

    /**
     * Constructor.
     * 
     * @param pId
     *            The transport ID
     */
    public AbstractTransport(final String pId) {
        super();

        mId = pId;
        mLabel = "?"; //$NON-NLS-1$
        mIncomingMessages = new LinkedBlockingQueue<Message>();
        mInvocationResults = Collections
                .synchronizedMap(new HashMap<Long, BlockingReference<ResponseMessage>>());
        mLastId = new AtomicLong(0);
        mActive = true;

        mLogger = new NemosLogger(AbstractTransport.class);
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.ITransport#close()
     */
    @Override
    public void close() {
        mActive = false;
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.ITransport#getId()
     */
    @Override
    public String getId() {
        return mId;
    }

    /**
     * Change the id.
     * 
     * @param pId
     *            The id to set
     */
    public void setId(final String pId) {
        mId = pId;
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.ITransport#receive(int,
     *      java.util.concurrent.TimeUnit)
     */
    @Override
    public Message receive(final int pTime, final TimeUnit pUnit) throws InterruptedException,
            RemoteException {
        if (isActive()) {
            return mIncomingMessages.poll(pTime, pUnit);
        }
        throw new RemoteException(mLogger.getTranslator().getMessage(RsaMessage.INACTIVE_TRANSPORT,
                getLabel()));
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.ITransport#call(nemos.remote.rsa.message.data.InvocationMessage)
     */
    @Override
    public ResponseMessage call(final InvocationMessage pCallMessage) throws RemoteException {
        try {
            return asyncCall(pCallMessage).get();
        } catch (final InterruptedException e) {
            final String lText = mLogger.getTranslator().getMessage(RsaMessage.INTERRUPTED_THREAD);
            throw new RemoteException(lText, e);
        }
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.ITransport#asyncCall(nemos.remote.rsa.message.data.InvocationMessage)
     */
    @Override
    public BlockingReference<ResponseMessage> asyncCall(final InvocationMessage pCallMessage)
            throws RemoteException {
        // Prepare message and reference
        pCallMessage.setId(mLastId.incrementAndGet());
        final BlockingReference<ResponseMessage> lResult = new BlockingReference<ResponseMessage>();
        final Long lId = Long.valueOf(pCallMessage.getId());
        mInvocationResults.put(lId, lResult);

        // Then send message
        mLogger.debug(RsaMessage.MESSAGE_OUT, getLabel(), pCallMessage);
        try {
            pCallMessage.prepareTransport();
        } catch (final IOException e) {
            throw new RemoteException(mLogger.getTranslator().getMessage(
                    RsaMessage.SERIALIZATION_ERROR, pCallMessage.toString()), e);
        }
        sendMessage(pCallMessage);
        return lResult;
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.ITransport#send(nemos.remote.rsa.message.data.Message)
     */
    @Override
    public void send(final Message pMessage) throws RemoteException {
        if (!(pMessage instanceof ResponseMessage)) {
            pMessage.setId(mLastId.incrementAndGet());
        }
        mLogger.debug(RsaMessage.MESSAGE_OUT, getLabel(), pMessage);
        try {
            pMessage.prepareTransport();
        } catch (final IOException e) {
            throw new RemoteException(mLogger.getTranslator().getMessage(
                    RsaMessage.SERIALIZATION_ERROR, pMessage.toString()), e);
        }
        sendMessage(pMessage);
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.ITransport#isActive()
     */
    @Override
    public boolean isActive() {
        return mActive;
    }

    /**
     * Send a message (ID has been set).
     * 
     * @param pMessage
     *            The message to send
     * @throws RemoteException
     *             If an error occurred
     */
    protected abstract void sendMessage(final Message pMessage) throws RemoteException;

    /**
     * Provides an incoming message to feed calls to the
     * {@link #receive(int, TimeUnit)} method.
     * 
     * @param pMessage
     *            The incoming message
     */
    protected void incomingMessage(final Message pMessage) {
        mLogger.debug(RsaMessage.MESSAGE_IN, getLabel(), pMessage);
        if (pMessage instanceof ResponseMessage) {
            final Long lId = Long.valueOf(pMessage.getId());
            final BlockingReference<ResponseMessage> lReference = mInvocationResults.remove(lId);
            if (lReference == null) {
                mLogger.error(RsaMessage.UNABLE_TO_SEND_RESPONSE, lId);
            } else {
                lReference.set((ResponseMessage) pMessage);
            }
        } else {
            mIncomingMessages.offer(pMessage);
        }
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.ITransport#setLabel(java.lang.String)
     */
    @Override
    public void setLabel(final String pLabel) {
        mLabel = pLabel;
    }

    /**
     * Get the label.
     * 
     * @return The label
     */
    protected String getLabel() {
        return mLabel;
    }
}
